package com.ydlclass.registry.zk;

import com.ydlclass.configuration.YrpcConfigurationHolder;
import com.ydlclass.core.ServerState;
import com.ydlclass.core.URL;
import com.ydlclass.registry.AbstractRegistry;
import com.ydlclass.registry.RegistryType;
import com.ydlclass.registry.util.ServiceHolder;
import com.ydlclass.serialization.json.JsonSerializer;
import com.ydlclass.utils.NetWorkUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @author it楠老师
 * @createTime 2023-03-09
 */
@Slf4j
public class ZookeeperRegistry extends AbstractRegistry {

    private ZooKeeper zooKeeper;

    /**
     * 创建zookeeper
     *
     * @return ZooKeeper
     */
    public ZooKeeper getZookeeper() {
        if (zooKeeper == null) {
            synchronized (ZookeeperRegistry.class) {
                if (zooKeeper == null) {
                    try {
                        // 1、连接zookeeper
                        CountDownLatch countDownLatch = new CountDownLatch(1);
                        zooKeeper = new ZooKeeper(url.getIpAndPortString(), 2000, new Watcher() {
                            @Override
                            public void process(WatchedEvent watchedEvent) {
                                Event.KeeperState state = watchedEvent.getState();
                                if (state == Event.KeeperState.SyncConnected) {
                                    if (log.isDebugEnabled()) {
                                        log.debug("zookeeper is connected!");
                                    }
                                } else if (state == Event.KeeperState.Disconnected) {
                                    if (log.isDebugEnabled()) {
                                        log.debug("zookeeper is disconnected!");
                                    }
                                }
                                countDownLatch.countDown();
                            }
                        });
                        countDownLatch.await();

                        Stat exists = zooKeeper.exists(servicePrefix.substring(0,servicePrefix.length()-1), null);
                        if(exists == null) {
                            zooKeeper.create(servicePrefix.substring(0,servicePrefix.length()-1),
                                    "Payload".getBytes(StandardCharsets.UTF_8),
                                    ZooDefs.Ids.OPEN_ACL_UNSAFE,
                                    CreateMode.PERSISTENT);
                        }

                    } catch (InterruptedException | IOException | KeeperException e) {
                        e.printStackTrace();
                    }
                }
            }

        }
        return zooKeeper;
    }

    public ZookeeperRegistry(URL url) {
        super(url);
        this.registryType = RegistryType.ZOOKEEPER;

    }

    /**
     * zookeeper的注册过程
     *
     * @param serviceName 服务名称
     */
    @Override
    public void doRegister(String serviceName) {
        try {
            // 1、获取zookeeper实例
            zooKeeper = getZookeeper();
            // 注册服务  临时节点   父-serviceName - 子：ip：port
            // 2、判断父节点是否存在
            Stat exists = zooKeeper.exists(serviceName, null);
            // 3、如果父节点不存在我们创建一个父节点
            if (exists == null) {
                zooKeeper.create(serviceName,
                        "Payload".getBytes(StandardCharsets.UTF_8),
                        ZooDefs.Ids.OPEN_ACL_UNSAFE,
                        CreateMode.PERSISTENT);
            }

            // 4、创建子目录，发布服务
            String targetPath = serviceName + "/" +
                    NetWorkUtils.getIp() + ":" + YrpcConfigurationHolder.get().getPort();
            zooKeeper.create(targetPath,
                    new JsonSerializer().serialize(new ServerState()),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL, (rc, path, ctx, name, stat) ->
                    {
                        if (log.isDebugEnabled()) {
                            log.debug("service [{}-{}-{}]  has been registered with the zookeeper!", serviceName, stat.getAversion(), targetPath);
                        }
                    }, YrpcConfigurationHolder.get().getTimeout());

        } catch (IOException | InterruptedException | KeeperException e) {
            log.error("A fault occurred while registry the service", e);
        }
    }

    @Override
    public void doUnregister(String serviceName) {
        zooKeeper = getZookeeper();
        try {
            zooKeeper.delete(serviceName + "/" + NetWorkUtils.getIp() + ":", 10,
                    (rc, path, ctx) -> {
                        if (log.isDebugEnabled()) {
                            log.debug("service [{}]  has been unregistered with the zookeeper!", path);
                        }
                    }, null
            );
        } catch (Exception e) {
            log.error("A fault occurred while deleting the service", e);
        }
    }

    /**
     * 查询服务列表
     *
     * @param serviceName 服务名称
     * @return 服务列表
     */
    @Override
    public List<InetSocketAddress> doLookup(String serviceName) {
        // 首先从缓存中查找
        List<InetSocketAddress> inetSocketAddresses = ServiceHolder.getInstance().getRemoteAddress(serviceName);
        if(inetSocketAddresses != null && inetSocketAddresses.size() > 0){
            return inetSocketAddresses;
        }

        // 未命中，重新拉取服务列表
        try {
            // todo 感知服务的动态上下线
            zooKeeper = getZookeeper();
            List<String> children = zooKeeper.getChildren(serviceName, true);
            inetSocketAddresses = resolveInetSocketAddress(children);
            ServiceHolder.getInstance().add(serviceName,inetSocketAddresses);
        } catch (KeeperException | InterruptedException e) {
            System.out.println(e.getMessage());
            log.error("A fault occurred while lookup the service", e);
        }
        return inetSocketAddresses;
    }

    private List<InetSocketAddress> resolveInetSocketAddress(List<String> children) {
        if (children == null || children.size() == 0) {
            throw new IllegalArgumentException("您应该指定一个合法的子节点集合");
        }

        return children.stream()
                .filter(addressString -> addressString.contains(":"))
                .map(addressString -> {
                    String[] hostAndPort = addressString.split(":");
                    return new InetSocketAddress(hostAndPort[0], Integer.valueOf(hostAndPort[1]));
                }).toList();
    }



}
